import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.streaming.api.collector.selector.OutputSelector;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.datastream.SplitStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

import java.util.ArrayList;
import java.util.List;


/**
 * @author wangzj
 * @description 分流练习
 * @date 2020/7/25 0:16
 */
public class SideOutPutDemo {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        //获取数据源
        List data = new ArrayList<Tuple3<Integer, Integer, Integer>>();
        data.add(new Tuple3<>(0, 1, 0));
        data.add(new Tuple3<>(0, 1, 1));
        data.add(new Tuple3<>(0, 2, 2));
        data.add(new Tuple3<>(0, 1, 3));
        data.add(new Tuple3<>(1, 2, 5));
        data.add(new Tuple3<>(1, 2, 9));
        data.add(new Tuple3<>(1, 2, 11));
        data.add(new Tuple3<>(1, 2, 13));

        //将列表数据转化成DataStream
        DataStreamSource<Tuple3<Integer, Integer, Integer>> items = env.fromCollection(data);

        /**
         * filter进行分流
         * 输入数据第一个元素为 0 的数据和第一个元素为 1 的数据
         * 分别被写入到了 zeroStream 和 oneStream 中
         */
        SingleOutputStreamOperator<Tuple3<Integer, Integer, Integer>> zeroStream = items
                .filter(value -> value.f0 == 0);
        SingleOutputStreamOperator<Tuple3<Integer, Integer, Integer>> oneStream = items
                .filter(value -> value.f0 == 1);
        zeroStream.printToErr();
        oneStream.print();

        /**
         * split进行流的切分
         */
        SplitStream<Tuple3<Integer, Integer, Integer>> splitStream = items
                .split(new OutputSelector<Tuple3<Integer, Integer, Integer>>() {
                    @Override
                    public Iterable<String> select(Tuple3<Integer, Integer, Integer> value) {
                        List<String> tags = new ArrayList<>();
                        if (value.f0 == 0) {
                            tags.add("zeroStream");
                        } else if (value.f0 == 1) {
                            tags.add("oneStream");
                        }
                        return tags;
                    }
                });
        splitStream.select("zeroStream").printToErr();
        splitStream.select("oneStream").print();

        /**
         * SideOutPut 分流
         */
        //1、定义 OutputTag
        OutputTag<Tuple3<Integer, Integer, Integer>> zeroStream1 = new OutputTag<Tuple3<Integer, Integer, Integer>>("zeroStream"){};
        OutputTag<Tuple3<Integer, Integer, Integer>> oneStream1 = new OutputTag<Tuple3<Integer, Integer, Integer>>("oneStream "){};
        //2、进行流的拆分和数据的写入
        SingleOutputStreamOperator<Tuple3<Integer, Integer, Integer>> processStream = items
                .process(new ProcessFunction<Tuple3<Integer, Integer, Integer>, Tuple3<Integer, Integer, Integer>>() {
                    @Override
                        public void processElement(Tuple3<Integer, Integer, Integer> value, Context ctx, Collector<Tuple3<Integer, Integer, Integer>> out) throws Exception {
                        if (value.f0 == 0) {
                            //将value的值写入zeroStream1中
                            ctx.output(zeroStream1, value);
                        } else if (value.f0 == 1) {
                            ctx.output(oneStream1, value);
                        }
                    }
                });
        //3、将数据转成DataStream
        DataStream<Tuple3<Integer, Integer, Integer>> zeroSideOutput = processStream.getSideOutput(zeroStream1);
        DataStream<Tuple3<Integer, Integer, Integer>> oneSideOutput = processStream.getSideOutput(oneStream1);
        zeroSideOutput.printToErr();
        oneSideOutput.print();

        env.execute("SideOutPut");
    }
}
